/**********************************************************************
 * FILE : Worker.java
 * CREATE DATE : 2011-11-24
 * DESCRIPTION :
 * CHANGE HISTORY LOG
 * ---------------------------------------------------------------------
 * NO.| DATE | NAME | REASON | DESCRIPTION
 * ---------------------------------------------------------------------
 * 1 | 2011-11-24 | Sting | 创建草稿版本
 * ---------------------------------------------------------------------
 */
package com.sgm.de.benchmark;

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Semaphore;

import javax.jms.JMSException;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.naming.Context;
import javax.naming.NamingException;

import com.sgm.dms.de.common.exceptions.DEException;
import com.sgm.dms.de.driver.IMessage;
import com.sgm.dms.de.driver.IQueue;
import com.sgm.dms.de.driver.IQueueManager;
import com.sgm.dms.de.driver.IMessage.MessageType;
import com.sgm.dms.de.driverimpl.ContextLocation;
import com.sgm.dms.de.driverimpl.JMSMessage;
import com.sgm.dms.de.driverimpl.JMSQueueManager;

/**
 * @author Sting
 *         <p>
 */
public class Worker implements Runnable {
    IQueueManager jmsMgr = null;
    Semaphore s = null;
    private Queue<IMessage> queue;
    public static volatile boolean toBeStop = false;
    private final ProcessContext pcontext;
    private Context ctx;
    private QueueConnectionFactory qconFactory;
    private QueueConnection qcon;
    private int threadSendLimit;

    /**
     * @param s
     * @param queue
     * @param config
     */
    public Worker(Semaphore s, Queue<IMessage> queue, ProcessContext pcontext) {
        this.s = s;
        this.queue = queue;
        this.pcontext = pcontext;
        this.threadSendLimit = pcontext.config.getThreadQuantitylimit();
        init();
    }


    /*
     * (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
        try {
            if (s != null)
                s.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            switch (pcontext.config.getWorktype()) {
                case send:
                    doSend();
                    break;
                case receive:
                    doReceive();
                    break;
                case syncsend:
                    doSyncSend();
                    break;
                case syncreceive:
                    doSyncReceive();
                    break;
                default:
                    break;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        if (s != null)
            s.release();
    }

    /**
     * TODO 新建方法
     * 
     * @return
     * @throws DEException
     */
    private void doSyncReceive() throws DEException {
        int sendsum = 0, recsum = 0;
        long sendKb = 0l, recKb = 0l;
        jmsMgr = (IQueueManager) new JMSQueueManager(ctx, qcon);
        String[] threadQueue = pcontext.queuePool.poll();
        String receiveQueueName = threadQueue[0];

        long lastExecTime = System.currentTimeMillis();
        IQueue receiveQueue = jmsMgr.accessQueue(receiveQueueName);

        // 有消息,或者processor没有停止标示,则继续运行
        while (!toBeStop) {
            if (recsum % 10 == 0) {
                long avgTime = (System.currentTimeMillis() - lastExecTime) / 10;
                System.out.printf(
                                "[worker %d ] <%s>receive sum:%d\treceive total:%d KB\tavg time:%d \t<SEND>send sum:%d\tsend total:%d\n",
                                Thread.currentThread().getId(), receiveQueueName, recsum, Math.round(recKb / 1024),
                                avgTime, sendsum,
                        Math.round(sendKb / 1024));
                lastExecTime = System.currentTimeMillis();
            }

            JMSMessage msg = new JMSMessage();
            receiveQueue.get(msg, 5000, false);
            if (msg.getDataLength() > 0 && msg.getReplyToQueueName() != null) {
                recKb += msg.getDataLength();
                recsum++;

                msg.setMessageType(MessageType.REPLY);
                msg.setCorrelationId(msg.getReplyToQueueManagerName().getBytes());
                String sendQueueName = msg.getReplyToQueueName();
                IQueue sendQueue = jmsMgr.accessQueue(sendQueueName);
                sendQueue.put(msg, false);
                sendsum++;
                sendKb += msg.getDataLength();
                msg = null;
            }
            if (threadSendLimit > 0 && recsum >= threadSendLimit)
                break;

        }
        System.out.printf(
                "[worker %d ] receive sum:%d\treceive total:%d KB \tsend sum:%d\tsend total:%d\n",
                Thread.currentThread().getId(), recsum, Math.round(recKb / 1024), sendsum, Math.round(sendKb / 1024));
        releaseResouce();
    }

    /**
     * @throws NamingException
     * @throws JMSException
     * @throws IllegalAccessException
     * @throws InstantiationException
     * @throws DEException
     */
    private void doSend() throws InstantiationException, IllegalAccessException, DEException {
        IMessage msg = null;
        int sum = 0;
        long totalkb = 0l;
        jmsMgr = (IQueueManager) new JMSQueueManager(ctx, qcon);
        String[] sendQueue = pcontext.queuePool.poll();
        long lastExecTime=System.currentTimeMillis();
        IQueue q = jmsMgr.accessQueue(sendQueue[0]);
        // 有消息,或者processor没有停止标示,则继续运行
        while ((msg = queue.poll()) != null || !toBeStop) {
            if (msg == null)
                continue;

            if (sum % 10 == 0){
                long avgTime = (System.currentTimeMillis() - lastExecTime) / 10;
                System.out.printf("[worker %d ] <%s>send sum:%d\tcommit total:%d KB\tavg time:%d\n",
                        Thread.currentThread().getId(), sendQueue, sum, Math.round(totalkb / 1024), avgTime);
                // 取睡眠时间,尽量靠近interval
                if (pcontext.config.getInterval() > 0 && avgTime < pcontext.config.getInterval()) {
                    try {
                        Thread.sleep(pcontext.config.getInterval() - avgTime);
                    } catch (InterruptedException e) {
                        System.err.println("worker sleep interruptered:" + e.getMessage());
                        continue;
                    }
                }
                lastExecTime = System.currentTimeMillis();
            }

            q.put(msg, true);
            jmsMgr.commit();
            totalkb += msg.getDataLength();
            sum++;
            msg = null;
            if (threadSendLimit > 0 && sum >= threadSendLimit)
                break;
        }
        System.out.printf("[worker %d summary] <%s>send sum:%d\ttotal:%dKB\n", Thread.currentThread().getId(),
                sendQueue, sum, Math.round(totalkb / 1024));
        releaseResouce();
    }


    /**
     * @param msg
     */
    private void doSyncSend() throws DEException {
        IMessage msg = null;
        int sendsum = 0, recsum = 0;
        long sendKb = 0l, recKb = 0l;
        jmsMgr = (IQueueManager) new JMSQueueManager(ctx, qcon);
        String[] threadQueue = pcontext.queuePool.poll();
        String sendQueueName = threadQueue[0];
        String receiveQueueName = threadQueue[1];
        long lastExecTime = System.currentTimeMillis();

        IQueue sendQueue = jmsMgr.accessQueue(sendQueueName);
        IQueue receiveQueue = jmsMgr.accessQueue(receiveQueueName);
        // 有消息,或者processor没有停止标示,则继续运行
        while ((msg = queue.poll()) != null || !toBeStop) {
            if (msg == null)
                continue;
            if (sendsum % 10 == 0) {
                long avgTime = (System.currentTimeMillis() - lastExecTime) / 10;
                System.out.printf(
                                "[worker %d ] <%s>send sum:%d\tsend total:%d KB\tavg time:%d\t <%s>receive sum:%d\treceive total%d\n",
                                Thread.currentThread().getId(), sendQueueName, sendsum, Math.round(sendKb / 1024),
                                avgTime, receiveQueueName, recsum, Math.round(recKb / 1024));
                // 取睡眠时间,尽量靠近interval
                if (pcontext.config.getInterval() > 0 && avgTime < pcontext.config.getInterval()) {
                    try {
                        Thread.sleep(pcontext.config.getInterval() - avgTime);
                    } catch (InterruptedException e) {
                        System.err.println("worker sleep interruptered:" + e.getMessage());
                        continue;
                    }
                }
                lastExecTime = System.currentTimeMillis();
            }
            String correlateId = Thread.currentThread().getId() + "-" + (sendsum + 1);
            msg.setReplyToQueueManagerName(correlateId);
            msg.setReplyToQueueName(receiveQueueName);
            sendQueue.put(msg, false);
            sendKb += msg.getDataLength();
            sendsum++;

            JMSMessage receiveMsg = new JMSMessage();
            receiveMsg.setCorrelationId(correlateId.getBytes());
            receiveQueue.get(receiveMsg, 8000, false);
            if (receiveMsg.getDataLength() > 0) {
                recKb += receiveMsg.getDataLength();
                recsum++;
            }
            receiveMsg = null;
            msg = null;
            if (threadSendLimit > 0 && recsum >= threadSendLimit)
                break;
        }
        System.out.printf(
                "[worker %d ] <%s>send sum:%d\tsend total:%dKB\t <%s>receive sum:%d\treceive total%d\n",
                Thread.currentThread().getId(), sendQueueName, sendsum, Math.round(sendKb / 1024), receiveQueueName,
                recsum,
                Math.round(recKb / 1024));
        releaseResouce();
    }

    /**
     * @param msg
     */
    private void doReceive() throws DEException {
        int sum = 0;
        long totalkb = 0l;
        jmsMgr = (IQueueManager) new JMSQueueManager(ctx, qcon);
        String queueName = pcontext.queuePool.poll()[0];
        long lastExecTime = System.currentTimeMillis();
        IQueue q = jmsMgr.accessQueue(queueName);
        // 有消息,或者processor没有停止标示,则继续运行
        while (!toBeStop) {
            if (sum % 10 == 0) {
                long avgTime = (System.currentTimeMillis() - lastExecTime) / 10;
                System.out.printf("[worker %d ] <%s>receive sum:%d\tcommit total:%d KB\tavg time:%d\n",
                        Thread.currentThread().getId(), queueName, sum, Math.round(totalkb / 1024), avgTime);
                lastExecTime = System.currentTimeMillis();
            }

            JMSMessage msg = new JMSMessage();
            q.get(msg, 5000, false);
            if (msg.getDataLength() > 0) {
                totalkb += msg.getDataLength();
                sum++;
                msg = null;
            }
            if (threadSendLimit > 0 && sum >= threadSendLimit)
                break;
        }
        System.out.printf("[worker %d summary] <%s>send sum:%d\ttotal:%dKB\n", Thread.currentThread().getId(),
                queueName, sum, Math.round(totalkb / 1024));
        releaseResouce();
    }


    private void releaseResouce() {
        jmsMgr.releaseResource();
    }

    /**
     * @return
     * @throws NamingException
     * @throws JMSException
     */
    public void init() {

        String url = "t3://" + pcontext.config.getJmsServerIp() + ":" + pcontext.config.getJmsServerPort();
        try {
            ctx = ContextLocation.getInitailContext(url, pcontext.config.getWlsUserName(),
                    pcontext.config.getWlsPassword());
            qconFactory = (QueueConnectionFactory) ctx.lookup(pcontext.config.getJmsFacName());
            qcon = qconFactory.createQueueConnection();
            System.out.println("[worker] initial context,get jms qcon" + qcon);
        } catch (NamingException e) {
            e.printStackTrace();
            System.exit(1);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    /*
     * @inheritDoc
     */
    @Override
    protected void finalize() throws Throwable {
        super.finalize();
        qcon.close();
        ctx.close();
    }

}
